package com.chis.jmdatatimer.yfjz;

import com.chis.jmdatatimer.comm.JedisConnect;
import com.chis.jmdatatimer.comm.PropertyConfigurer;
import com.chis.jmdatatimer.comm.ThreadPoolSingle;
import com.chis.jmdatatimer.yfjz.threads.KafkaSrvThread;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.common.TopicPartition;
import redis.clients.jedis.ShardedJedis;

import java.util.Properties;


/**
 * @author wlj
 * @version 2019年01月10日
 * @Description
 * @Company zwx
 */
public class SrcKafkaSrvTask extends KafkaConsumerTemplate {

    @Override
    public void run() {
        Properties properties = initKafkaData();
        int num = Integer.valueOf(PropertyConfigurer.getProperty("num.partitions"));
        ShardedJedis jedis = null;
        try {
            jedis = JedisConnect.getInstance().jedisPool.getResource();
            jedis.select(0);
            for(int part = 0 ; part < num ; part++){
                Long offset = null;
                String key = getGroupId()+"-"+getTopic()+"-"+part;
                String topicpart = jedis.get(key);
                if(StringUtils.isBlank(topicpart)){
                    //主题分区不存在
                    offset = 0L;
                }else{
                    //主题分区存在
                    offset =Long.valueOf(topicpart);
                }
                //开线程
                KafkaSrvThread thread = new KafkaSrvThread(properties,new TopicPartition(getTopic(), part),offset,getGroupId());
                ThreadPoolSingle.getInstance().getPool().execute(thread);
            }
        } catch (Exception e){
            e.printStackTrace();
        } finally {
            if(jedis != null){
                jedis.close();
            }
        }

    }

    @Override
    public String getGroupId() {
        return "SRC_GROUP_SRV";
    }

    @Override
    public String getTopic() {
        String citytag = null;
        try {
            citytag = PropertyConfigurer.getProperty("spark.kafkatopic.citytag");
        } catch (Exception e) {
            e.printStackTrace();
        }
        return "TOPIC_SRV_"+citytag;
    }

}
